package com.hulk.rpc.invoke;

import com.google.common.base.Strings;
import com.google.common.collect.Maps;
import com.google.common.collect.Streams;
import com.hulk.rpc.annotation.HulkService;
import com.hulk.rpc.common.HulkConnectService;
import com.hulk.rpc.server.HulkConnectServiceImpl;
import com.hulk.rpc.utils.FastMap;
import com.hulk.rpc.utils.concurrent.ConcurrentArrayList;
import com.hulk.rpc.utils.concurrent.ConcurrentIntToObjectArrayMap;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;

import java.beans.BeanInfo;
import java.beans.Introspector;
import java.beans.PropertyDescriptor;
import java.lang.reflect.Method;
import java.lang.reflect.Modifier;
import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
import java.util.stream.Stream;

/**
 * server本地方法调用(javassist)
 *
 * @author xuxukang
 * @date 2019-03-09 22:27
 */
@Slf4j
@Getter
public class ServiceInvokeFactory {

  private final String group;
  private final String app;
  private final String restPrefix;

  /**
   * rpc服务调用缓存
   */
  private final ConcurrentArrayList<JavassistInvoker<?>> rpcInvokerMap = new ConcurrentArrayList<>();
  /**
   * rest服务调用缓存
   */
  private volatile FastMap<String, JavassistInvoker<?>> restInvokerMap = new FastMap<>(32, 0.5F);
  /**
   * 用于判断server中的哪些类已经加载了
   */
  private final ConcurrentMap<String, Boolean> classRegisterMap = Maps.newConcurrentMap();
  /**
   * 服务方法名缓存
   */
  private final ConcurrentIntToObjectArrayMap<String> serviceMethodNameMap = new ConcurrentIntToObjectArrayMap<>();

  private final AtomicInteger classIdGenerator = new AtomicInteger();
  private final ConcurrentMap<String, Integer> classIdMap = Maps.newConcurrentMap();

  public ServiceInvokeFactory(String group, String app) {
    this.group = group;
    this.app = app;
    this.restPrefix = "/" + group + "/" + app + "/";

    HulkConnectServiceImpl hulkConnectService = new HulkConnectServiceImpl(this);
    register(HulkConnectService.class, hulkConnectService);
  }

  @SuppressWarnings("unchecked")
  public <T> Invoker<T> get(int serviceId) {
    return (Invoker<T>) rpcInvokerMap.get(serviceId);
  }

  /**
   * 通过restPath获取invoker
   *
   * @param restPath
   */
  @SuppressWarnings("unchecked")
  public <T> Invoker<T> get(String restPath) {
    return (Invoker<T>) restInvokerMap.get(restPath);
  }

  public synchronized void register(Class<?> clazz, Object service) {
    register(Map.of(clazz, service));
  }

  public synchronized void register(Map<Class<?>, Object> map) {
    if (map == null) {
      throw new InvokeException("map is null");
    }

    final AtomicInteger serviceIdCounter = new AtomicInteger(rpcInvokerMap.size());

    List<JavassistInvoker<?>> invokers = map//
            .entrySet()//
            .stream()//
            .flatMap(kv -> convertToInvokerStream(kv.getKey(), kv.getValue(), serviceIdCounter))//
            .collect(Collectors.toList());

    if (invokers.isEmpty()) {
      return;
    }

    rpcInvokerMap.addAll(invokers);
    putRestInvoker(invokers);
  }

  /**
   * 获取已注册的class
   */
  public List<String> getClassRegisterList() {
    return new ArrayList<>(classRegisterMap.keySet());
  }

  /**
   * 获取已注册的method:serviceId映射
   */
  public Map<String, Integer> getMethodRegisterMap() {
    Map<String, Integer> registerMap = new HashMap<>();

    for (int i = 0; i < rpcInvokerMap.size(); i++) {
      JavassistInvoker<?> invoker = rpcInvokerMap.get(i);
      Method method = invoker.method;

      registerMap.put(InvokerUtils.getServiceMethodName(group, app, method), i);
    }

    return registerMap;
  }

  /**
   * 获取已注册的rest服务列表
   */
  public List<String> getRestRegisterList() {
    return Streams//
            .stream((Iterable<String>) restInvokerMap.keys())//
            .map(path -> restPrefix + path)//
            .collect(Collectors.toList());
  }

  public String getServiceMethodName(int serviceId) {
    String serviceMethodName = serviceMethodNameMap.get(serviceId);

    if (serviceMethodName != null) {
      return serviceMethodName;
    }

    return serviceMethodNameMap.getOrUpdate(serviceId, () -> {
      Method method = get(serviceId).getMethod();
      return InvokerUtils.getServiceMethodName(group, app, method);
    });
  }

  private synchronized void putRestInvoker(List<JavassistInvoker<?>> invokers) {
    FastMap<String, JavassistInvoker<?>> map = new FastMap<>(restInvokerMap);

    invokers.forEach(invoker -> {
      String restPath = InvokerUtils.getRestPath(invoker.method);

      if (!Strings.isNullOrEmpty(restPath)) {
        map.put(restPath, invoker);

        if (log.isInfoEnabled()) {
          log.info(InvokerUtils.getServiceMethodName(group, app, invoker.method) + " restPath:"
                  + restPrefix + restPath);
        }
      }
    });

    restInvokerMap = map;
  }

  private <T> Stream<JavassistInvoker<T>> convertToInvokerStream(Class<?> clazz, Object service,
                                                                 AtomicInteger serviceIdCounter) {

    if (clazz == null) {
      throw new InvokeException("clazz cannot be null");
    }

    if (service == null) {
      throw new InvokeException("service cannot be null");
    }

    if (!clazz.isInstance(service)) {
      throw new InvokeException("clazz is the interface, service is the implemention");
    }

    if (!clazz.isInterface()) {
      throw new InvokeException("the clazz must be interface");
    }

    if (!Modifier.isPublic(clazz.getModifiers())) {
      throw new InvokeException("the clazz must be public");
    }

    HulkService classConfig = clazz.getAnnotation(HulkService.class);
    if (classConfig != null && classConfig.ignore()) {
      if (log.isInfoEnabled()) {
        log.info(clazz + " ignore");
      }

      return Stream.empty();
    }

    if (classRegisterMap.containsKey(InvokerUtils.getServiceClassName(group, app, clazz))) {
      return Stream.empty();
    }

    Method[] allMethods = clazz.getMethods();
    for (Method method : allMethods) {
      if (!CompletableFuture.class.equals(method.getReturnType())) {
        throw new RuntimeException("method return-type must be CompletableFuture, "
                + InvokerUtils.getServiceMethodName(group, app, method));
      }
    }

    classRegisterMap.put(InvokerUtils.getServiceClassName(group, app, clazz), Boolean.TRUE);
    if (log.isInfoEnabled()) {
      log.info("service " + clazz + " register");
    }

    Stream<Method> methodStream = Stream//
            .of(allMethods)//
            .filter(m -> Modifier.isPublic(m.getModifiers()))//
            .filter(m -> !Modifier.isStatic(m.getModifiers()))//
            .peek(m -> {
              if (!CompletableFuture.class.equals(m.getReturnType())) {
                throw new RuntimeException("method return-type must be CompletableFuture, "
                        + InvokerUtils.getServiceMethodName(group, app, m));
              }
            })//
            .filter(m -> {
              HulkService methodConfig = m.getAnnotation(HulkService.class);

              if (methodConfig != null && methodConfig.ignore()) {
                if (log.isInfoEnabled()) {
                  log.info(InvokerUtils.getServiceMethodName(group, app, m) + " ignore");
                  return false;
                }
              }

              return true;
            });

    // 特殊处理，保证编号正确性
    if (HulkConnectService.class.equals(clazz)) {
      methodStream = methodStream.sorted((m0, m1) -> {
        int sort0 = HulkConnectService.serviceOrderMap.get(m0.getName());
        int sort1 = HulkConnectService.serviceOrderMap.get(m1.getName());

        return sort0 - sort1;
      });
    }

    // 注册classId
    for (Method method : allMethods) {
      registerClassId(method);
    }

    return methodStream//
            .map(m -> {
              int serviceId = serviceIdCounter.getAndIncrement();

              if (log.isInfoEnabled()) {
                log.info(InvokerUtils.getServiceMethodName(group, app, m) + " serviceId:" + serviceId);
              }

              return new JavassistInvoker<>(serviceId, service, m.getDeclaringClass(), m);
            });
  }

  private void registerClassId(Method method) {
    if (method == null) {
      return;
    }

    Type returnType = method.getGenericReturnType();
    registerClassId(returnType);

    Type[] genericParameterTypes = method.getGenericParameterTypes();
    for (Type genericParameterType : genericParameterTypes) {
      registerClassId(genericParameterType);
    }
  }

  private void registerClassId(Type type) {
    if (type == null) {
      return;
    }

    if (type instanceof ParameterizedType) {
      ParameterizedType parameterizedType = (ParameterizedType) type;
      Type rawType = parameterizedType.getRawType();
      registerClassId(rawType);

      Type[] actualTypeArguments = parameterizedType.getActualTypeArguments();
      for (Type actualTypeArgument : actualTypeArguments) {
        registerClassId(actualTypeArgument);
      }
    } else if (type instanceof Class) {
      Class<?> clazz = (Class<?>) type;

      if (clazz.isInterface()) {
        return;
      }

      if (clazz.isPrimitive()) {
        return;
      }

      if (clazz.isArray()) {
        return;
      }

      if (clazz.isEnum()) {
        return;
      }

      if (clazz.isAnnotation()) {
        return;
      }

      if (clazz.isAnonymousClass()) {
        return;
      }

      if (clazz.equals(CompletableFuture.class)) {
        return;
      }

      String className = clazz.getName();

      if (className.startsWith("java.lang.")) {
        return;
      }

      if (classIdMap.containsKey(className)) {
        return;
      }

      int classId = classIdMap//
              .computeIfAbsent(className, key -> classIdGenerator.getAndIncrement());

      log.info("register Serializer.classId " + className + ":" + classId);

      Method[] childMethods = getChildMethods(clazz);
      for (Method childMethod : childMethods) {
        registerClassId(childMethod);
      }
    }
  }

  private Method[] getChildMethods(Class<?> clazz) {
    BeanInfo info;
    try {
      info = Introspector.getBeanInfo(clazz);
    } catch (Throwable t) {
      return new Method[0];
    }

    PropertyDescriptor[] descriptors = info.getPropertyDescriptors();
    if (descriptors.length == 0) {
      return new Method[0];
    }

    Method[] childMethods = new Method[descriptors.length << 1];

    for (int i = 0; i < descriptors.length; i++) {
      PropertyDescriptor descriptor = descriptors[i];
      childMethods[i] = descriptor.getReadMethod();
      childMethods[i << 1] = descriptor.getWriteMethod();
    }

    return childMethods;
  }
}
